package com.cskaoyan.order.mq;

import com.cskaoyan.mall.commons.exception.BizException;
import com.cskaoyan.mall.pay.dto.PayOrderCancelRequest;
import com.cskaoyan.mall.pay.dto.PayOrderCancelResponse;
import com.cskaoyan.order.api.PayApi;
import com.cskaoyan.order.constant.OrderConstants;
import com.cskaoyan.order.dal.entitys.Order;
import com.cskaoyan.order.dal.persistence.OrderMapper;
import com.cskaoyan.order.dto.CancelOrderRequest;
import com.cskaoyan.order.service.OrderCoreService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.io.UnsupportedEncodingException;
import java.nio.charset.StandardCharsets;
import java.util.List;

/**
 * @author: 牟治宇
 * @description 订单延迟消费者
 * @date: 2022/8/2 17:22
 */
@Component
@Slf4j
public class DelayOrderCancelConsumer {

    private DefaultMQPushConsumer pushConsumer;

    @Autowired
    OrderMapper orderMapper;

    @Autowired
    OrderCoreService orderCoreService;

    @Autowired
    PayApi payApi;
    @PostConstruct
    public void init() {

        pushConsumer = new DefaultMQPushConsumer("delay_order_consumer");
        pushConsumer.setNamesrvAddr("127.0.0.1:9876");
        pushConsumer.setMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {

                MessageExt message = list.get(0);
                byte[] body = message.getBody();
                try {
                    String orderId = new String(body, 0, body.length, StandardCharsets.UTF_8);
                    Order order = orderMapper.selectByPrimaryKey(orderId);
                    Integer status = order.getStatus();

                    if (status == OrderConstants.ORDER_STATUS_INIT) {
                        PayOrderCancelRequest payOrderCancelRequest = new PayOrderCancelRequest();
                        payOrderCancelRequest.setOrderId(orderId);
                        PayOrderCancelResponse payOrderCancelResponse = payApi.payOrderCancel(payOrderCancelRequest);
                        log.info("订单已被系统取消");
                        //TODO 这里返回的是boolean?
                        //如果返回
                        if (Integer.parseInt(payOrderCancelResponse.getCode()) ==4 ){
                            CancelOrderRequest cancelOrderRequest = new CancelOrderRequest();
                            cancelOrderRequest.setOrderId(orderId);
                            orderCoreService.cancelOrder(cancelOrderRequest);
                            log.info("订单已成功取消");
                        }
                    }

                } catch (Exception e) {
                    throw new BizException("1008612580", "订单取消失败");
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        try {
            pushConsumer.subscribe("delay_order_cancel", "*");
            pushConsumer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }
}
